package com.example.config;

import com.example.handler.ReceiverMessageHandler;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import javax.annotation.Resource;

/**
 * mqtt入站
 *
 * @author ysb
 * @version V1.0
 * @Date 2024/11/12
 **/
@Configuration
public class MqttInboundConfiguration {

    @Resource
    private MqttConfigurationProperties mqttConfigurationProperties;
    @Resource
    private MqttPahoClientFactory mqttPahoClientFactory;
    @Resource
    private ReceiverMessageHandler receiverMessageHandler;
    @Resource
    private MqttPahoMessageDrivenChannelAdapter messageProducer;

    // 消息通道
    @Bean
    public MessageChannel mqttInboundChannel() {
        return new DirectChannel();
    }

    // 配置入站适配器，作用：设置订阅主题，以及指定消息的相关属性
    @Bean
    public MessageProducer messageProducer() {
        MqttPahoMessageDrivenChannelAdapter mqttPahoMessageDrivenChannelAdapter = new MqttPahoMessageDrivenChannelAdapter(
                mqttConfigurationProperties.getUrl(),
                mqttConfigurationProperties.getSubClientId(),
                mqttPahoClientFactory,
                mqttConfigurationProperties.getSubTopic().split(",")
        );

        mqttPahoMessageDrivenChannelAdapter.setQos(1);
        mqttPahoMessageDrivenChannelAdapter.setConverter(new DefaultPahoMessageConverter());
        mqttPahoMessageDrivenChannelAdapter.setOutputChannel(mqttInboundChannel());
        return mqttPahoMessageDrivenChannelAdapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInboundChannel")
    public MessageHandler messageHandler() {
        return receiverMessageHandler;
    }

    /**
     * 动态添加订阅主题
     *
     * @param topic 新的订阅主题
     * @throws MqttException 如果订阅失败
     */
    public void addSubscription(String topic, int qos) throws MqttException {
        if (messageProducer != null) {
            messageProducer.addTopic(topic, qos);
        } else {
            throw new IllegalStateException("MQTT adapter is not initialized");
        }
    }
}
